package com.cmby.flinkbrick

import java.util

import com.cmby.flinkbrick.dto.{ConfUtils, JobUtils}
import com.cmby.flinkbrick.entity.{JobInfo, JobResult}
import com.cmby.flinkbrick.helper.SQLHelper
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object FlinkBrick {



  def main(args: Array[String]): Unit = {
    // 1.flink环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val streamSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    val streamTableEnv = StreamTableEnvironment.create(streamEnv, streamSettings)

    // 2.配置加载
    val jobConfInputStream = this.getClass.getClassLoader.getResourceAsStream("conf/kafka2mysql.yml")
    val jobResult: JobResult = JobUtils.parseJob(jobConfInputStream)
    val expression = jobResult.getJobInfo.getExpression

//    val connectorSQL= SQLHelper.buildConnectorSQL(jobResult.getJobInfo.getSourceList.get(0).params)
//    SQLHelper.buildColumnSQL();


    val createInputSQL = "CREATE TABLE user_behavior_source (\n  " +
      "`user_id` BIGINT,\n  " +
      "`price` DOUBLE,\n  " +
      "`quantity` DOUBLE,\n  " +
      "`cost` AS price * quanitity,\n  " +
      ") WITH (\n  " +
      "'connector' = 'kafka',\n  " +
      "'topic' = 'user_behavior',\n " +
      "'properties.bootstrap.servers' = 'localhost:9092',\n  " +
      "'properties.group.id' = 'userGroup',\n  " +
      "'scan.startup.mode' = 'earliest-offset',\n  " +
      "'format' = 'json'" +
      "...\n);"
    streamTableEnv.executeSql(createInputSQL)

    val createOutputSQL = "CREATE TABLE user_behavior_sink (\n  " +
      "`user_id` BIGINT,\n  " +
      "`price` DOUBLE,\n  " +
      "`quantity` DOUBLE,\n  " +
      "`cost` AS price * quanitity,\n  " +
      ") WITH (\n  " +
      "'connector' = 'kafka',\n  " +
      "'topic' = 'user_behavior',\n " +
      "'properties.bootstrap.servers' = 'localhost:9092',\n  " +
      "'properties.group.id' = 'userGroup',\n  " +
      "'scan.startup.mode' = 'earliest-offset',\n  " +
      "'format' = 'json'" +
      "...\n);"
    streamTableEnv.executeSql(createOutputSQL)

    streamTableEnv.sqlQuery(expression)

    val result = streamTableEnv.sqlQuery("select * from user_behavior")
    result.executeInsert("user_behavior_sink")

  }
}
